package cn.xiaobai.spring.boot.listener;

import cn.xiaobai.spring.boot.annotaion.KafkaTopicsSubscriber;
import cn.xiaobai.spring.boot.utils.ProxyUtils;
import lombok.Getter;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.core.env.Environment;
import org.springframework.kafka.listener.BatchConsumerAwareMessageListener;
import org.springframework.kafka.support.TopicPartitionOffset;
import org.springframework.util.CollectionUtils;
import org.springframework.util.ReflectionUtils;

import java.lang.reflect.Method;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * kafka 批量消息监听器
 * 并静态代理自定义注解
 *
 * @Author yangdaji
 * @Date 2023/11/3
 * @PackageName pattern.kafka.listener
 * @ClassName BatchKafkaListener
 */
public class BatchKafkaListener implements BatchConsumerAwareMessageListener<Object, Object>, ApplicationContextAware {

    private transient Map<String, Map<Object, Method>> topicMessagesMap = new ConcurrentHashMap<>();

    @Getter
    private TopicPartitionOffset[] topicPartitionOffsets;


    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        Environment environment = applicationContext.getEnvironment();
        subscriber(applicationContext, environment);
    }

    private void subscriber(ApplicationContext applicationContext, Environment environment) {
        Map<String, Object> beansWithAnnotation = applicationContext.getBeansWithAnnotation(KafkaTopicsSubscriber.class);
        List<TopicPartitionOffset> topicList = new LinkedList<>();
        beansWithAnnotation.values().forEach(sub -> {
            KafkaTopicsSubscriber annotation = AnnotationUtils.findAnnotation(sub.getClass(), KafkaTopicsSubscriber.class);
            String[] topics = annotation.topics();
            int partition = annotation.partition();
            for (String topic : topics) {
                String top = environment.resolvePlaceholders(topic);
                topicList.add(new TopicPartitionOffset(top, partition));
                // 绑定代理对象
                ProxyUtils.bindProxy(sub, top, topicMessagesMap);
            }
        });
        topicPartitionOffsets = topicList.toArray(new TopicPartitionOffset[0]);
    }


    @Override
    public void onMessage(List<ConsumerRecord<Object, Object>> data, Consumer<?, ?> consumer) {
        for (ConsumerRecord<Object, Object> datum : data) {
            String topic = datum.topic();
            Object message = datum.value();
            consumer.groupMetadata().groupId();
            Map<Object, Method> objectMethodMap = topicMessagesMap.get(topic);
            if (!CollectionUtils.isEmpty(objectMethodMap)) {
                objectMethodMap.forEach((proxy, method) -> {
                    ReflectionUtils.invokeMethod(method, proxy, message);
                });
            }
            consumer.commitAsync();
        }

    }


}
